package com.zishuimuyu.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;

/**
 * @author 向佩环
 * @description mqtt连接类
 * @company 北京木联能软件股份有限公司
 * @date 2022/9/7
 */
@Slf4j
public class MyMqttClient {

    /*自定义监听，多线程异步解析数据**/
    private static MqttClient client;

    /*mqtt数据接收、回调**/
    @Autowired
    private MyMqttCallback myMqttCallback;

    /*自定义监听，多线程异步解析数据**/
    @Autowired
    private MqttMessageListener mqttMessageListener;

    public static MqttClient getClient() {
        return client;
    }

    public static void setClient(MqttClient client) {
        MyMqttClient.client = client;
    }


    public void setMyMqttCallback(MyMqttCallback myMqttCallback) {
        this.myMqttCallback = myMqttCallback;
    }

    /**
     * 设置mqtt连接参数
     *
     * @param username
     * @param password
     * @param willTopic
     * @param willContent
     * @param completionTimeout
     * @param keepAliveInterval
     *
     * @return
     */
    public MqttConnectOptions setMqttConnectOptions(String username, String password, String willTopic, String willContent, int completionTimeout, int keepAliveInterval) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(completionTimeout);
        options.setKeepAliveInterval(keepAliveInterval);
        options.setCleanSession(true);
        options.setWill(willTopic, willContent.getBytes(), 2, false);
        return options;
    }

    /**
     * 连接mqtt服务端，得到MqttClient连接对象
     */
    public void connect(List<String> urls, String username, String password, String clientId, String willTopic, String willContent, Integer completionTimeout, Integer keepAliveInterval, String dataTopic) throws MqttException {
        if (null == client) {
            MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, willTopic, willContent, completionTimeout, keepAliveInterval);
            boolean connected = false;
            mqttConnectOptions.setCleanSession(true);
            for (String url : urls) {
                try {
                    MyMqttClient.setClient(new MqttClient(url, clientId, new MemoryPersistence()));
                    myMqttCallback.setClientAndOptions(client, mqttConnectOptions);
                    client.setCallback(myMqttCallback);

                    if (!client.isConnected()) {
                        client.connect(mqttConnectOptions);
                    } else {
                        client.disconnect();
                        client.connect(mqttConnectOptions);
                    }
                    log.info("MQTT 成功连接 URL: " + url);
                    connected = true;
                    break;
                } catch (MqttException e) {
                    log.warn("无法连接 URL: " + url, e);
                }
            }

            if (!connected) {
                log.error("无法连接任何配置的 URL");
            }
        }
    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param pushMessage
     * @param topic
     */
    public void publish(String pushMessage, String topic) {
        publish(pushMessage, topic, 0, false);
    }
    /**
     * 发布，默认qos为0，非持久化
     *
     * @param pushMessage
     * @param topic
     */
    public void publish(MqttMessage pushMessage, String topic) throws MqttException {
        client.publish(topic, pushMessage);
    }

    /**
     * 发布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(String pushMessage, String topic, int qos, boolean retained) {
        MqttMessage message = new MqttMessage();
        message.setPayload(pushMessage.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        MqttTopic mqttTopic = client.getTopic(topic);
        if (null == mqttTopic) {
            log.error("topic is not exist");
        }
        MqttDeliveryToken token;//Delivery:配送
        synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁
            try {
                token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
                token.waitForCompletion(1000L);
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }
    /**
     * 发布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(MqttMessage pushMessage, String topic, int qos, boolean retained) {
        MqttMessage message = new MqttMessage();
        message.setPayload(pushMessage.getPayload());
        message.setQos(qos);
        message.setRetained(retained);
        MqttTopic mqttTopic = client.getTopic(topic);
        if (null == mqttTopic) {
            log.error("topic is not exist");
        }
        MqttDeliveryToken token;//Delivery:配送
        synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁
            try {
                token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
                token.waitForCompletion(1000L);
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 订阅某个主题
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            client.subscribe(topic, qos, mqttMessageListener);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅某个主题
     *
     * @param topic
     */
    public void subscribe(String topic) {
        try {
            client.subscribe(topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 取消订阅主题
     *
     * @param topic 主题名称
     */
    public void cleanTopic(String topic) {
        if (client != null && client.isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            log.info("取消订阅失败！");
        }
    }

}
